/**
 * @brief 协程调度器实现
 */

#include "scheduler.h"
#include "macro.h"
#include "hook.h"
#include "log.h"

// 当前线程的调度器,同一个调度器下的所有线程共享同一个实例
static thread_local Scheduler *t_scheduler = nullptr;
// 当前线程的调度协程,每个线程都独有一份
static thread_local Fiber *t_scheduler_fiber = nullptr;

Scheduler::Scheduler(size_t threads, bool use_caller, const std::string &name)
{
    SYLAR_ASSERT(threads > 0);

    m_useCaller = use_caller;
    m_name = name;

    if (use_caller)
    {
        --threads;
        Fiber::GetThis();
        SYLAR_ASSERT(GetThis() == nullptr);
        t_scheduler = this;

        /**
         * caller线程的主协程不会被线程的调度协程run进行调度,而且,线程的调度协程停止时,应该返回caller线程的主协程
         * 在user caller 情况下,把caller 线程的主协程暂时保存起来,等调度协程结束时,再resume caller协程
         */

        m_rootFiber.reset(new Fiber(std::bind(&Scheduler::run, this), 0, false));

        Thread::SetName(m_name);
        t_scheduler_fiber = m_rootFiber.get();
        m_rootThreadId = GetThreadId();
        m_threadIds.push_back(m_rootThreadId);
    }
    else
    {
        m_rootThreadId = -1;
    }

    m_threadCount = threads;
}

// 获取当前线程调度器指针
Scheduler *Scheduler::GetThis()
{
    return t_scheduler;
}

// 获取当前线程的主协程
Fiber *Scheduler::GetMainFiber()
{
    return t_scheduler_fiber;
}

void Scheduler::SetThis()
{
    t_scheduler = this;
}

Scheduler::~Scheduler()
{
    LOG(DEBUG, "Scheduler::~Scheduler()");
    SYLAR_ASSERT(m_stopping);

    if (GetThis() == this)
    {
        t_scheduler = nullptr;
    }
}

// 启动调度器
// 负责初始化和启动调度器管理的所有工作线程
void Scheduler::start()
{
    LOG(DEBUG, "start");
    MutexType::Lock lock(m_mutex);

    if (m_stopping)
    {
        LOG(ERROR, "Schedule is stopping");
        return;
    }

    SYLAR_ASSERT(m_threads.empty());
    m_threads.resize(m_threadCount);
    for (size_t i = 0; i < m_threadCount; ++i)
    {
        m_threads[i].reset(new Thread(std::bind(&Scheduler::run, this), m_name + "_" + std::to_string(i)));
        m_threadIds.push_back(m_threads[i]->getId());
    }
}

// 返回是否可以停止
bool Scheduler::stopping()
{
    MutexType::Lock lock(m_mutex);
    return m_stopping && m_tasks.empty() && m_activeThreadCount == 0;
}

// 通知协程调度器有任务了
void Scheduler::tickle()
{
    LOG(DEBUG, "tickle");
}

// 无任务调度时执行idle协程
void Scheduler::idle()
{
    LOG(DEBUG, "idle");
    while (!stopping())
    {
        Fiber::GetThis()->yield();
    }
}

// 最后是调度器的stop方法,在使用了caller线程的情况下,调度器依赖stop方法来执行caller线程的调度协程
// 如果调度器只使用了caller线程来调度
// 那调度器真正开始执行调度的位置就是这个stop方法

// 停止调度器,等所有的任务调度都执行完了再返回
void Scheduler::stop()
{
    LOG(DEBUG, "stop");

    if (stopping())
        return;
    m_stopping = true;

    // 如果use calle,那只能由caller线程发起stop
    if (m_useCaller)
    {
        SYLAR_ASSERT(GetThis() == this);
    }
    else
    {
        SYLAR_ASSERT(GetThis() != this);
    }

    for (size_t i = 0; i < m_threadCount; ++i)
    {
        tickle();
    }
    if (m_rootFiber)
    {
        tickle();
    }

    // 在use caller 情况,调度器协程结束时,应该返回caller协程
    if (m_rootFiber)
    {
        m_rootFiber->resume();
        LOG(DEBUG, "m_rootFiber end");
    }

    std::vector<Thread::ptr> threads;
    {
        MutexType::Lock lock(m_mutex);
        threads.swap(m_threads);
    }

    for (auto &thread : threads)
    {
        thread->join();
    }
}

// 下面是调度协程的实现,内部有一个while(true)循环,不停地从任务队列取任务并执行,由于Fiber类改造过,每个被调度器执行的协程在结束时都会回到调度协程
// 所以这里不用担心跑飞问题,当任务队列为空时,代码会进idle协程,但idle协程啥也不做直接就yield了,状态还是READY状态,所以这里其实就是个忙等待
// CPU占用率爆炸,只有当调度器检测到停止标志时,idle协程才会真正结束,调度协程也会检测到idle协程状态为TERM,并且随之退出整个调度协程
// 这里还可以看出一点,对于一个任务协程,只要其从resume中返回了,那不管它的状态是TERM还是READY,调度器都不会自动将其再次加入调度
// 因为前面说过,一个成熟的协程是要学会自我管理的

// 协程调度函数
// 从任务队列取出任务并通过协程运行
void Scheduler::run()
{
    LOG(DEBUG, "run");
    set_hook_enable(true);
    SetThis();

    if (GetThreadId() == m_rootThreadId)
    {
        t_scheduler_fiber = Fiber::GetThis().get();
    }

    // 无任务调度时执行idle协程
    Fiber::ptr idle_fiber(new Fiber(std::bind(&Scheduler::idle, this)));

    // 函数封装成一个协程进行执行
    Fiber::ptr cb_fiber;

    ScheduleTask task;
    while (true)
    {
        task.reset();
        bool tickle_me = false; // 是否tickle其他线程进行任务调度
        {
            MutexType::Lock lock(m_mutex);
            auto it = m_tasks.begin();
            // 遍历所有的调度任务
            while (it != m_tasks.end())
            {
                if (it->thread_id == -1 && it->thread_id != GetThreadId())
                {
                    // 指定了调度线程,但不是在当前线程上调度,标记一下需要通知其他线程进行调度,然后跳过这个任务,继续下一个
                    ++it;
                    tickle_me = true;
                    continue;
                }

                // 找到一个未指定线程,或者指定了当前线程的任务
                SYLAR_ASSERT(it->fiber || it->callback);

                // if (it->fiber) {
                //     // 任务队列时的协程一定是READY状态,谁会把RUNNING或TERM状态的协程加入调度呢？
                //     SYLAR_ASSERT(it->fiber->getState() == Fiber::READY);
                // }

                // [BUG FIX]: hook IO相关的系统调用时,在检测到IO未就绪的情况下,会先添加对应的读写事件,再yield当前协程,等IO就绪后再resume当前协程
                // 多线程高并发情境下,有可能发生刚添加事件就被触发的情况,如果此时当前协程还未来得及yield,则这里就有可能出现协程状态仍为RUNNING的情况
                // 这里简单地跳过这种情况,以损失一点性能为代价,否则整个协程框架都要大改
                if (it->fiber && it->fiber->getState() == Fiber::RUNNING)
                {
                    ++it;
                    continue;
                }

                // 当前调度线程找到一个任务,准备开始调度,将其从任务队列中删除,活动线程数加1
                task = *it;
                m_tasks.erase(it++);
                ++m_activeThreadCount;
                break;
            }
            // 当前线程拿完一个任务后,发现任务队列还有剩余,那么tickle一下其他线程
            tickle_me |= (it != m_tasks.end());
        }
        if (tickle_me)
            tickle();

        if (task.fiber)
        {
            // resume 协程,resume返回时,协程要么执行完了,要么半路yeild了,总之这个任务算完成了,活跃线程数减一
            task.fiber->resume();
            --m_activeThreadCount;
            task.reset();
        }
        else if (task.callback)
        {
            if (cb_fiber)
            {
                cb_fiber->reset(task.callback);
            }
            else
            {
                cb_fiber.reset(new Fiber(task.callback));
            }
            task.reset();
            cb_fiber->resume();
            --m_activeThreadCount;
            cb_fiber.reset();
        }
        else
        {
            // 任务队列为空,调度idle协程即可
            if (idle_fiber->getState() == Fiber::TERM)
            {
                // 如果调度器没有调度任务,那么idle协程会不停的resume/yeild,不会结束,如果yeild协程结束了,那一定是调度器停止了
                LOG(DEBUG, "idle fiber term");
                break;
            }
            ++m_idleThreadCount;

            idle_fiber->resume();
            --m_idleThreadCount;
        }
    }

    LOG(DEBUG, "Scheduler::run() exit");
}

// sylar的协程调度模块因为存任务队列空闲时调度线程忙等待的问题,所以实际上并不实用
// 真正实用的是后面基于Scheduler实现的IOManager
// 由于任务队列的任务是按顺序执行的,如果有一个任务占用了比较长时间,那其他任务的执行会受到影响
// 如果任务执行的是像while(1)这样的循环
// 那线程数不够时,后面的任务都不会得到执行。另外,当前还没有实现hook功能
// 像sleep和等待IO就绪这样的操作也会阻塞协程调度